package com.shujia.flink

import com.shujia.flink.Demo7StreamTableApi.Event
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.types.Row
import org.apache.flink.table.api.Tumble

object Demo8TableWindows {
  def main(args: Array[String]): Unit = {
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    //设置时间类型为事件事件
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val tableEnv = StreamTableEnvironment.create(streamEnv)


    val ds = streamEnv

      /**
        * a,1
        * a,2
        */
      .socketTextStream("node1", 7777)
      .map(_.split(","))
      .map(split => Event(split(0), split(1).toInt, split(2).toLong))
      .assignAscendingTimestamps(_.ts) //指定那一列作为事件时间


    val table = tableEnv.fromDataStream(ds, 'word, 'num, 'ts.rowtime)


    table
      .window(Tumble over 5.seconds on 'ts as 'w)
      .groupBy('w, 'word)
      .select('word, 'num.sum)
      .toAppendStream[Row]
      .print()

    streamEnv.execute()


  }

  case class Event(word: String, num: Int, ts: Long)

}
